package com.viknix.threadpool.manager.config.dump.engine;

import com.viknix.threadpool.manager.config.dump.processor.TpmTaskProcessor;
import com.viknix.threadpool.manager.server.dependence.exception.ServerProcessException;
import com.viknix.threadpool.manager.server.dependence.factory.ExecutorFactory;
import com.viknix.threadpool.manager.server.dependence.factory.NameThreadFactory;
import com.viknix.threadpool.manager.server.dependence.task.AbstractDelayTask;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: Dongqi
 * @Date: 2021/11/20 12:53
 * @Version 1.0
 * @Description: 定期处理集群间通信任务的引擎
 */
@Slf4j
public class TpmDelayTaskExecuteEngine extends AbstractTpmTaskExecuteEngine<AbstractDelayTask> {

    /**
     * 该线程池为单线程
     */
    private final ScheduledExecutorService processingExecutor;

    /**
     * 用于存放 DumpTask：重新从数据库拉去配置的任务 {@link ProcessRunnable#run}
     * key:dataId + group + tenant + isBeta + tag ; value:DumpTask
     */
    protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;

    protected final ReentrantLock lock = new ReentrantLock();

    public TpmDelayTaskExecuteEngine(String name, long processInterval) {
        this(name, 32, processInterval);
    }

    public TpmDelayTaskExecuteEngine(String name, int initCapacity, long processInterval) {
        tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity);
        processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));
        // 定期执行集群结点之间同步实例数据任务 processInterval = 100L
        processingExecutor
                .scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);
    }

    @Override
    public int size() {
        lock.lock();
        try {
            return tasks.size();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public boolean isEmpty() {
        lock.lock();
        try {
            return tasks.isEmpty();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public AbstractDelayTask removeTask(Object key) {
        lock.lock();
        try {
            AbstractDelayTask task = tasks.get(key);
            if (null != task && task.shouldProcess()) {
                return tasks.remove(key);
            } else {
                return null;
            }
        } finally {
            lock.unlock();
        }
    }

    @Override
    public Collection<Object> getAllTaskKeys() {
        Collection<Object> keys = new HashSet<Object>();
        lock.lock();
        try {
            keys.addAll(tasks.keySet());
        } finally {
            lock.unlock();
        }
        return keys;
    }

    @Override
    public void shutdown() throws ServerProcessException {
        processingExecutor.shutdown();
    }

    @Override
    public void addTask(Object key, AbstractDelayTask newTask) {
        lock.lock();
        try {
            AbstractDelayTask existTask = tasks.get(key);
            if (null != existTask) {
                newTask.merge(existTask);
            }
            tasks.put(key, newTask);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 刷新内存中线程池的配置信息
     */
    protected void processTasks() {
        // 获取所有需要处理的task的key：dataId + group + tenant + isBeta + tag
        Collection<Object> keys = getAllTaskKeys();
        for (Object taskKey : keys) {
            // 获取需要执行的 DumpTask
            AbstractDelayTask task = removeTask(taskKey);
            if (null == task) {
                continue;
            }
            // DumpProcessor
            TpmTaskProcessor processor = getProcessor(taskKey);
            if (null == processor) {
                log.error("processor not found for task, so discarded. " + task);
                continue;
            }
            try {
                // 执行 DumpTask
                if (!processor.process(task)) {
                    // 如果执行失败，则重新添加任务 到队列中
                    retryFailedTask(taskKey, task);
                }
            } catch (Throwable e) {
                log.error("Tpm task execute error : " + e.toString(), e);
                // 如果执行失败，则重新添加任务 到队列中
                retryFailedTask(taskKey, task);
            }
        }
    }

    private void retryFailedTask(Object key, AbstractDelayTask task) {
        task.setLastProcessTime(System.currentTimeMillis());
        addTask(key, task);
    }

    private class ProcessRunnable implements Runnable {

        @Override
        public void run() {
            try {
                processTasks();
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }
        }
    }
}
